跳到主要内容

庞大规模的数据流引擎（Large-Scale Dataflow Engines）

由 Peter Bailis 介绍

被选中的论文​

Jeff Dean and Sanjay Ghemawat. MapReduce: Simplified Data Processing on Large Clusters. OSDI, 2004.

Yuan Yu, Michael Isard, Dennis Fetterly, Mihai Budiu. DryadLINQ: A System for General-Purpose Distributed Data-Parallel Computing Using a High-Level Language. OSDI, 2008.

回首过去十年，数据管理领域发展不息，MapReduce和其随后的大规模数据处理系统无疑是最具有颠覆性与争议性的。廉价的商业存储与稳步增长的数据容量，使得不少互联网供应商摒弃了传统的数据库系统和数据仓库系统，转而根据自己的需要，构建自定义的引擎。谷歌围绕大规模系统所推出的一系列产品，涵盖谷歌文件系统[^10]，MapReduce，Chubby[^6]以及 BigTable[^7]，在市场里面最负盛名与最具影响。而在几乎所有的情况下面，这些新自主研发的系统，都对传统数据库中包括高（抽象）层次编程语言，查询优化，以及极具效率的执行策略在内的一小部分的特性进行了实现。由此，这些系统以及随后到来的开源 Hadoop 生态系统，在许多开发者当中变得流行起来。这就使得许多投资，市场，研究热点以及开发同时倾注到这些平台之中。不过截至目前，这些平台依旧处于不断的变化之中，同时作为一个生态系统，开始向着类传统数据仓库演变 --- 并且做了很多重要的调整。我们将在这里，就这些趋势进行反思。

历史与继任者（History and Successors）​

我们所选择的第一篇读物，是Google在2004年发布的 MapReduce 。这是一个为处理谷歌规模分布式数据，而构建出来的简化版本并行查询分布式库，着重关注从抓取的页面当中重建数据索引。而在（MapReduce诞生的）那个时候，传统数据仓库并不能处理这种程度的工作负载。

而相较于传统的数据仓库，MapReduce提供了一个相当低层次的接口（两阶段数据库流，two-stage dataflow），这个接口与容错性密切相关（在两阶段数据流中终止物化操作）。同样重要的事情在于，MapReduce 被设计为一个并行编程函数库，而不是端到端的数据仓库解决方案；举例来说，MapReduce 依托了谷歌的文件系统来构建自身的存储。而在当时，数据库社区的许多成员，纷纷谴责这种架构，简单，毫无效率，以及应用场景狭隘[^8]。

尽管原版的MapReduce发布于2003年，但直到2006年之前，与MapReduce有关的Google外部活动依旧非常低迷。而在2006年，伴随着雅虎的开源MapReduce实现方案Hadoop推出。外部世界的兴趣迅速聚焦而来：一年之内，包括Dryad(微软公司)[^15]，Hive(脸书公司)[^26]，Pig(雅虎公司)[^22]等一系列项目都启动了研发。这些系统，我们都称之为后MapReduce系统，它们在开发群体当中魅力非凡 --- 它们主要集中于硅谷 --- 伴随着巨量的风险投资。跨系统、数据库以及网络社区，调查研究了包括调度计划，排队者缓解（straggler mitigation），容错，以及用户自定义查询优化，以及可替代编程模型等在内的诸多议题。

到了最近，后MapReuce时代的系统们对它们的接口以及功能展开了包括复杂性声明接口，查询优化策略，高性能运行时在内的诸多拓展工作。并在今天实现了越来越多的传统关系型数据库功能集。新一代的数据处理引擎，如Spark[^27]，F1[^24]，Impala[^16]，Tex[^1]，Naiad[^21]，Flink/Stratosphere[^2]，AsterixDB[^3]，以及Drill[^14]已经构建出高层次的诸如SQL这样的查询语言来，同时赋予了它们处理通用图数据操作符，索引，以及其它可用的结构化数据输入源的能力。在Hadoop生态系统中，数据流引擎已经成为了如SQL[^4],[^26]，图数据处理[^12],[^19]，机器学习[^11],[^25]在内的一系列高级功能以及声明式接口的基石。

人们对于流数据的处理也是越发兴趣蓬勃，业界重新审视了2000年之后在数据库社区里面提出来的许多前沿概念。而许多成长中的商业社区与开源社区都对结构化的，以及半结构化的数据输入源的“连接器”（connectors），目录功能（如HCatalog），数据服务和有限事务处理能力（如HBase）做了开发。尽管这些社区框架的许多功能（如查询优化器）相较于不少成熟的商用数据库，依旧显得初级，但是其进步照样神速。

DryadLINQ是我们在这个篇章中，选择好的第二篇阅读材料，它最为有趣的地方可能就是它的接口：它是一组为数据处理而准备好的嵌入式语言绑定工具，集成了微软公司的.NET LINQ 接口，提供了并行化的搜集数据库。DryadLINQ通过早期的 Dryad 系统[^15]来执行查询，依托基于重放的容错机制为运行时抽象数据流图实现了运行时。当 DryadLINQ 依旧限制程序员们展开一些无副作用的数据集转换工作（包括“类SQL”操作符），但是它依旧提供出了一个抽象层次远高于MapReduce的接口。DryadLINQ 的语言独立性（与具体编程语言无关），轻量级容错性，以及基本的查询优化技术对随后包括 Apache Spark[^27]，微软 Naiad[^21]在内的数据流系统，发挥了重要的影响。

影响与遗产​

很难再次发生的 MapReduce 现象至少给我们今天的世界，带来了至少三个延续至今的影响。这些情况 --- 就如同分布式数据流本身 --- 不一定新颖，但是后MapReduce时代的系统已经广泛地拓展了它们的影响：

  1. 模式的灵活性
    这可能就是最为重要的影响，传统的数据流仓库系统，是盖上了墙的花园：它们所输入的数据，是原始的，精心策划过的，以及体系化的。而相对的，MapReduce 系统提供的是抽象好的结构化数据，而无论其处于干净状态，还是混沌状态（whether clean or dirty）,精心加工过，还是没有处理过（curated or not）。它们的加载顺序，并不需要遵循一定的步骤。这也就意味着，用户可以在思考拿数据去做什么之前，就把它们存储进来。而当我们与实际情况比较之后，这种类型的存储（比如，Hadoop 的文件系统之中）就比传统的数据仓库更为廉价，因为用户存储数据的时间，可以长得多。这也是导致用户从传统的数据仓库迁徙出来，以及“大数据”迅速取得增长的一个关键性因素。不断增长着的存储格式（比如，Avro，Parquet，RCFile），将半结构化的数据同那些先进前沿的，诸如列布局相结合起来。相对于 XML，这种新的半结构化的数据的浪潮，更加具有灵活性。作为一个结果，提取-转换-装载（extract-transform-load，ETL）成为了后MapReduce引擎的主要工作负载。在各个层级上面，无论是数据分析师，到程序员，到供应商，无论如何去强调模式灵活性都不为过，因为我们深信在未来这将会变得更加重要。不过，这种多元性代价并不是完全免费的：策划类似于“数据湖”（data lakes）的行为非常昂贵（甚至远远多于存储），而我们将会在第十二章中就这个话题展开详细的讨论。
  2. 接口的灵活性
    今天，绝大多数的用户，都与类SQL语言的大数据引擎打交道。不过，这些数据引擎同样允许用户使用范例组合展开编程。举例来说，一个组织可能会在同一个框架之内，使用命令行代码来展开文件解析的工作，使用 SQL 来对列执行投影的操作，并且使用机器学习的子模块对结果展开聚类分析的工作。而在 DryadLINQ 中，紧密的，惯用的语言接口司空见惯，这也进一步提升了可编程性。而与此同时，历史上面的一些传统数据库引擎，也开始尝试支持用户自定义计算操作符，以简化表达，同时它使得传统的类似于SQL的传统关系型设施同用户自定义计算结果，更易于集成起来。接口的灵活性和集成性，对于数据分析套件而言，是一项强壮的卖点；这种能力集成了 ETL，数据分析，以及对于程序员而言，非常容易的在单个系统框架内的预处理流程 --- 但是对于那些传统的 GI 工具的作者而言，这些事情没有什么必要性，它们依旧使用着传统的 JDBC 接口。
  3. 架构的灵活性
    一种常见的对于关系型数据库的批评就在于，它们的架构联系地过于紧密：存储，查询处理，内存管理，事务处理，以及诸如此类的组件，交织在一起，同时在实践之中，它们又缺少了清晰明确的接口。而与之相对的，作为自下而上的一个结果的造物，Hadoop 生态系统有效地把数据仓库，以一系列组件的形式，构造出来。到了今天，组织可以就在原始的文件系统（如HDFS）上面编写并且运行程序，同时许许多多的数据流引擎（如Spark），就会使用一些更为前沿的数据分析包（如GraphLab[^18]，Parameter Server[^17]），或者类SQL（如Impala[^16]）。这种灵活性提升了性能的负载，并且促进了前所未有的混合匹配组件与分析包的融合。这种架构上面的灵活性，也可能就是那些系统构建者和供应商最感兴趣的地方，因为它们可以在这种自由性的直到下面，构建它们自己的付费基础设施。

简单来说，当今分布式数据管理系统的一个主题，就是灵活性与异质性：围绕着存储的格式，围绕着计算的范例，以及系统的实现。在这些情况的指导之下，存储格式的异构性可能会带来一个乃至于更多个数量级的影响，简单来说，因为它对于新手，专家，以及系统的架构师，都存在着影响。而与之相对的，计算范例更多地影响的是专家与架构师，而系统的实现，则更多地影响架构师。这三种影响对于数据库领域的研究者而言，实际上相互关联，同时它们也将一直伴随着市场的影响力与生命力而存在。

对未来的展望​

从理性的角度来看待，MapReduce 非常短命，但是其极致的架构（为后来的软件作者）打开了设计的空间。这种架构，非常简单，而且易于伸缩，而它在开源领域的成功，令相当多的人意识到，市场，对于那些能够顺利替换的，同时具备价格灵活性质的解决方案，存在着需求（更不用说，那些基于开源之上的廉价数据仓库解决方案，它们是核心因素）。而由此引发出来的兴趣，依旧令相当多的人感到惊讶，而这也是由多方面的因素所导致出来的一个结果，比如社区时代的精神，聪明的市场营销手法，经济生态，以及技术的变迁。而除此之外，尝试了解这些崭新的系统与关系型数据库之间，究竟哪些是根本性改进，哪些是工程上面的改进，也是非常有意思的事情。

到了今天，对于合适恰当的大规模伸缩性数据处理流程的争论依旧是进行时。作为案例，Rasmussen 等人为为什么中间件的容错性在超级大规模（100+ node）集群的情况下并非必要，提供了强有力的证据[^23]。而另外一个案例，McSherry 等人，则生动地说明了许多工作负载，可以在单独的服务器（或者线程里面）进行有效的处理，进而打消了分布式的必要性[^20]。而到了最近，类似于 GraphLab 这样的项目[^18]认为，性能对于特定领域的系统是相当必要的。而随即展开的工作，包括Grail[^9]与GraphX[^12]，则提出分歧意见，它们认为这种需要并不契合于场景。而最近所提出来的另外一个浪潮，同样则在流数据处理，图数据处理，异步程序编程，通用性机器学习上面，提出了有关的建议。而这些东西真的是特定领域的系统所需要的吗？又或者，存在着一种分析引擎，可以都把它们调动起来？时间将会证明一切，但就我个人看来，我感觉一切正在走向融合。

最后，倘若如果我们不对 Spark 展开一个讨论，那么我们就难免太失职了，尽管 Spark 只有六岁大小，但是它依旧在开发者群体当中变得风靡，并且它得到了风险创投基金（比如，Databricks）的大力支持，且同诸如 Cloudera 与 IBM 这样的大型企业，建立了深度的合作关系。正当我们因为 DryadINQ 的历史影响力与技术深度，将它引入进红皮书，作为后 MapReduce 系统时代的案例的时候，Spark的论文[^27]，在项目工作的早些时候，写就出来了，并且在最近做了包括 SparkSQL[^4]在内的拓展，它们同时值得一读。就像 Hadoop 一样，Spark在早期的成熟阶段，就已经吸引了很多外部的关注。而到了今天，Spark在其特性达到与传统数据仓库一样成熟的阶段之前，依旧有着很长的道路要走。不过，它的特性迅速完善，而对于让Spark成为类似于MapReduce之于Hadoop生态下这样的成功者的呼声也非常高昂。举个例子，Cloudera正在计划在Hadoop生态下面用Spark替换Hadoop[^13]。时间将会告诉我们，这些期待是否准确。而与此同时，传统数据仓库同后MapReduce时代的系统间的差距，正在快速缩小，这就使得新的这些系统，将会在达到与传统时代系统一个成熟度的基础上(as good)，青出于蓝而胜于蓝（but also much more）。

由 Michael Stonebraker 做出来的技术评论​

2015年10月26日

最近，人们对数据分析这个围绕着“大数据”市场热词下的领域产生了极大的兴趣。从历史上看，这个名词的内涵就是商业智能数据分析，由商务智能应用程序（Cognos，Business Objects 这样的对象）和传统的关系型数据仓库（如 Teredata，Vertica，Red Shift，Greenplum）这样的组合来联合提供服务。而又到了最近，它们又同“数据科学”（data science）相结合起来。在这个大的背景下面，我们可以先对十年前的Map-Reduce做一个回顾，它由谷歌公司制造，用于支持它们的网站爬虫数据。而（Map-Reduce提出来）之后，市场上的人们对其的基本观感是：“聪明睿智的谷歌公司研究出了 Map-Reduce，它将成为谷歌的下一桩大事，它必定很好”（Google is smart; Map-Reduce is Google’s next big thing, so it must be good）。而Cloudera，Hortonworks以及Facebook成为了宣传 Map-Reduce（而它们就研究出了类似于Hadoop这样的开源产品）的主力军。

而数年以前，市场上还依旧被 Map-Reduce 灌着蛊惑的苦酒（koolaid）。就在这个时候，谷歌公司停止为它们的应用程序使用本为满足它们的需求而研究的Map-Reduce，转向了 BigTable。而到了大约五年之后，全世界的其它人才知道谷歌公司早就发掘出来的结论。Map-Reduce并不是一个具有广阔伸缩性能力的架构。

实际上，Map-Reduce 的惨败主要源自于下面的两个问题：

  1. 它作为一个平台，并不适合于构建数据仓库。
    任何的商业数据库产品的内部接口，都没有看起来像 Map-Reduce 的，这是有充分的理由的。因此，数据库并不期待这一类型的平台。
  2. 对于那些希望构建分布式应用程序的人而言，这个平台同样不合适
    这不仅仅是因为 Map-Reduce 的接口对于分布式应用程序而言，并不成熟，同时还有，它的消息传递系统是基于文件系统的，速度如此之慢，以至于让人提不起兴趣来。

当然，这些原因并没有阻挡那些 Map-Reduce 供应商的步伐。它们只是简单地将它们的平台重新命名为 HDFS（这是一种文件系统），同时它们基于 HDFS 来构建一些不涵盖 Map-Reduce 的应用程序。举例来说，Cloudera 最近引入了 Impala，它是一种 SQL 引擎，但是却并非架构于 Map-Reduce 之上。而实际上，Impala 也并不真正使用 HDFS，而是选择直接跳过这一层，用 Linux 的文件系统来直接展开数据读写的工作。HortonWorks和Facebook下面的项目也采取了类似的方式。就此，Map-Reduce的使用群体已经转向了SQL的群体，同时Map-Reduce作为一层接口，也随即成为历史。当然，HDFS在经由SQL引擎使用的时候，存在严重的问题，因此我们并不清楚它最终能否真正被应用起来，这还有待观察（it is not clear that it will have any legs, but that remains to be seen）。不过无论如何，Map-Reducs-HDFS 的市场最终都将同 SQL 数据仓库市场合并。同时可能会成为最为流行的系统解决方案。简单来说，作为一款分布式数据库平台，Map-Reduce已经失败，而它的供应商们，则把 HDFS 作为一款数据仓库产品下面的文件系统而得以应用。

这就把我们带到了 Spark 之上。在最初的时候，Spark被认为是一个快速版本的Map-Reduce。因为它有着一个基于内存的快速消息传输接口。因此，当它为应用程序提供服务的时候，它并不存在严重的性能问题。不过，按照 Spark 首席作者（lead author）Mater Zaharia 的说法来看，超过70%的用户都通过SparkSQL来实现对于Spark的访问。实际上，Spark被当作一款SQL引擎来进行使用，而并不是一款分布式应用平台！在这种背景下面，Spark就会存在定位不清的问题。如果它是一款SQL平台，它就需要一定的机制，来保障持久性，索引，以及用户间的主存数据分布，元数据目录等事情，以此才可以在 SQL 的数据仓库领域保持竞争能力。这样，就使得 Spark 将会向着数据仓库平台来转型，就像 Hadoop 曾经走过的道路那样。

而从另外一个方面来看，那30%并不经由SparkSQL来访问Spark的用户，主要源自于Scala。而这条道路，则可能是一条分布式计算的道路。在这种环境下面，Spark是一款负责可靠的分布式计算平台。不过，这里依旧存在一些需要思考的事情。首先，一般情况下面，数据科学家将会把数据的管理与数据的分析结合起来。较高的性能往往来自于两者的组合。在Spark中，则没有这样的组合，因为Spark的数据格式一般在这两个任务中不一定通用。第二，Spark是一款只在内存中工作的数据库（至少现在是这样）。伸缩性的要求可能会在未来的某个时间，让其修订这个问题。在这种场景下，观察Spark的未来进化方向，将会是一件有意思的事情。

简单来说，我希望提供下面的这些格言：

  1. 谷歌认定的好主意不一定就是你需要采用的主意
  2. 不要相信所有的市场营销，而一定要看看对应产品实际可以带来的好处。这应当尤其适用于那些性能方面的宣传
  3. 程序员组成的社区中，人们对“下一代亮点产品”情有独钟。而这可能会对你的组织带来混乱，因为这些亮点产品的半衰期可能会相当短

参考​

[1] Apache Tez. https://tez.apache.org/.
[2] A. Alexandrov, R. Bergmann, S. Ewen, J.-C. Freytag, F. Hueske, A. Heise, O. Kao, M. Leich, U. Leser, V. Markl, et al. The Stratosphere platform for big data analytics. The VLDB Journal, 23(6):939–964, 2014.
[3] S. Alsubaiee, Y. Altowim, H. Altwaijry, A. Behm, V. Borkar, Y. Bu, M. Carey, I. Cetindil, M. Cheelangi, K. Faraaz, et al. Asterixdb: A scalable, open source bdms. In VLDB, 2014.
[4] M. Armbrust, R. S. Xin, C. Lian, Y. Huai, D. Liu, J. K. Bradley, X. Meng, T. Kaftan, M. J. Franklin, A. Ghodsi, et al. Spark SQL: Relational data processing in spark. In SIGMOD, 2015.
[5] S. Babu and H. Herodotou. Massively parallel databases and MapReduce systems. Foundations and Trends in Databases, 5(1):1–104, 2013.
[6] M. Burrows. The chubby lock service for loosely-coupled distributed systems. In OSDI, 2006.
[7] F. Chang, J. Dean, S. Ghemawat, W. C. Hsieh, D. A. Wallach, M. Burrows, T. Chandra, A. Fikes, and R. E. Gruber. Bigtable: A distributed storage system for structured data. In OSDI, 2006.
[8] D. DeWitt and M. Stonebraker. Mapreduce: A major step backwards. The Database Column, 2008.
[9] J. Fan, A. Gerald, S. Raj, and J. M. Patel. The case against specialized graph analytics engines. In CIDR, 2015.
[10] S. Ghemawat, H. Gobioff, and S.-T. Leung. The google file system. In SOSP, 2003.
[11] A. Ghoting, R. Krishnamurthy, E. Pednault, B. Reinwald, V. Sindhwani, S. Tatikonda, Y. Tian, and S. Vaithyanathan. Systemml: Declarative machine learning on mapreduce. In ICDE, 2011.
[12] J. E. Gonzales, R. S. Xin, D. Crankshaw, A. Dave, M. J. Franklin, and I. Stoica. Graphx: Unifying data-parallel and graph-parallel analytics. In OSDI, 2014.
[13] D. Harris. Forbes: Why Cloudera is saying ’Goodbye, MapReduce’ and ’Hello, Spark’, 2015. http://fortune.com/ 2015/09/09/cloudera-spark-mapreduce/.
[14] M. Hausenblas and J. Nadeau. Apache Drill: Interactive ad-hoc analysis at scale. Big Data, 1(2):100–104, 2013.
[15] M. Isard, M. Budiu, Y. Yu, A. Birrell, and D. Fetterly. Dryad: distributed data-parallel programs from sequential building blocks. In EuroSys, 2007.
[16] M. Kornacker, A. Behm, V. Bittorf, T. Bobrovytsky, C. Ching, A. Choi, J. Erickson, M. Grund, D. Hecht, M. Jacobs, et al. Impala: A modern, open-source sql engine for hadoop. In CIDR, 2015.
[17] M. Li, D. G. Andersen, J. W. Park, A. J. Smola, A. Ahmed, V. Josifovski, J. Long, E. J. Shekita, and B.-Y. Su. Scaling distributed machine learning with the parameter server. In OSDI, 2014
[18] Y. Low, D. Bickson, J. Gonzalez, C. Guestrin, A. Kyrola, and J. M. Hellerstein. Distributed graphlab: a framework for machine learning and data mining in the cloud. In VLDB, 2012.
[19] G. Malewicz, M. H. Austern, A. J. Bik, J. C. Dehnert, I. Horn, N. Leiser, and G. Czajkowski. Pregel: a system for large-scale graph processing. In SIGMOD, 2010.
[20] F. McSherry, M. Isard, and D. G. Murray. Scalability! But at what COST? In HotOS, 2015.
[21] D. G. Murray, F. McSherry, R. Isaacs, M. Isard, P. Barham, and M. Abadi. Naiad: A timely dataflow system. In SOSP, 2013.
[22] C. Olston, B. Reed, U. Srivastava, R. Kumar, and A. Tomkins. Pig latin: a not-so-foreign language for data processing. In SIGMOD, 2008.
[23] A. Rasmussen, V. T. Lam, M. Conley, G. Porter, R. Kapoor, and A. Vahdat. Themis: An i/o-efficient mapreduce. In SoCC, 2012.
[24] J. Shute, R. Vingralek, B. Samwel, B. Handy, C. Whipkey, E. Rollins, M. Oancea, K. Littlefield, D. Menestrina, S. Ellner, et al. F1: A distributed sql database that scales. In VLDB, 2013.
[25] E. R. Sparks, A. Talwalkar, V. Smith, J. Kottalam, X. Pan, J. Gonzalez, M. J. Franklin, M. Jordan, T. Kraska, et al. Mli: An api for distributed machine learning. In ICDM, 2013.
[26] A. Thusoo, J. S. Sarma, N. Jain, Z. Shao, P. Chakka, S. Anthony, H. Liu, P. Wyckoff, and R. Murthy. Hive: A warehousing solution over a map-reduce framework. In VLDB, 2009.
[27] M. Zaharia, M. Chowdhury, T. Das, A. Dave, J. Ma, M. McCauley, M. J. Franklin, S. Shenker, and I. Stoica. Resilient distributed datasets: A fault-tolerant abstraction for in-memory cluster computing. In NSDI, 2012.